// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <nebula/io/slow.h>

#include <algorithm>
#include <cstring>
#include <mutex>
#include <random>
#include <thread>
#include <utility>

#include <nebula/core/buffer.h>
#include <nebula/io/util_internal.h>

#include <turbo/utility/status.h>
#include <turbo/files/io_util.h>
#include <turbo/log/logging.h>
#include <turbo/random/random.h>
namespace nebula::io {

    // Multiply the average by this ratio to get the intended standard deviation
    static constexpr double kStandardDeviationRatio = 0.1;

    class LatencyGeneratorImpl : public LatencyGenerator {
    public:
        ~LatencyGeneratorImpl() override = default;

        LatencyGeneratorImpl(double average_latency, int32_t seed)
                : gen_(static_cast<decltype(gen_)::result_type>(seed)),
                  latency_dist_(average_latency, average_latency * kStandardDeviationRatio) {}

        double NextLatency() override {
            // std::random distributions are unlikely to be thread-safe, and
            // a RandomAccessFile may be called from multiple threads
            std::lock_guard<std::mutex> lock(mutex_);
            return std::max<double>(0.0, latency_dist_(gen_));
        }

    private:
        std::default_random_engine gen_;
        std::normal_distribution<double> latency_dist_;
        std::mutex mutex_;
    };

    LatencyGenerator::~LatencyGenerator() {}

    void LatencyGenerator::Sleep() {
        std::this_thread::sleep_for(std::chrono::duration<double>(NextLatency()));
    }

    std::shared_ptr<LatencyGenerator> LatencyGenerator::create(double average_latency) {
        return std::make_shared<LatencyGeneratorImpl>(
                average_latency, static_cast<int32_t>(turbo::get_random_seed()));
    }

    std::shared_ptr<LatencyGenerator> LatencyGenerator::create(double average_latency,
                                                             int32_t seed) {
        return std::make_shared<LatencyGeneratorImpl>(average_latency, seed);
    }

    //////////////////////////////////////////////////////////////////////////
    // SlowInputStream implementation

    SlowInputStream::~SlowInputStream() { internal::CloseFromDestructor(this); }

    turbo::Status SlowInputStream::close() { return stream_->close(); }

    turbo::Status SlowInputStream::abort() { return stream_->abort(); }

    bool SlowInputStream::closed() const { return stream_->closed(); }

    turbo::Result<int64_t> SlowInputStream::tell() const { return stream_->tell(); }

    turbo::Result<int64_t> SlowInputStream::read(int64_t nbytes, void *out) {
        latencies_->Sleep();
        return stream_->read(nbytes, out);
    }

    turbo::Result<std::shared_ptr<Buffer>> SlowInputStream::read(int64_t nbytes) {
        latencies_->Sleep();
        return stream_->read(nbytes);
    }

    turbo::Result<std::string_view> SlowInputStream::peek(int64_t nbytes) {
        return stream_->peek(nbytes);
    }

    //////////////////////////////////////////////////////////////////////////
    // SlowRandomAccessFile implementation

    SlowRandomAccessFile::~SlowRandomAccessFile() { internal::CloseFromDestructor(this); }

    turbo::Status SlowRandomAccessFile::close() { return stream_->close(); }

    turbo::Status SlowRandomAccessFile::abort() { return stream_->abort(); }

    bool SlowRandomAccessFile::closed() const { return stream_->closed(); }

    turbo::Result<int64_t> SlowRandomAccessFile::get_size() { return stream_->get_size(); }

    turbo::Status SlowRandomAccessFile::Seek(int64_t position) { return stream_->Seek(position); }

    turbo::Result<int64_t> SlowRandomAccessFile::tell() const { return stream_->tell(); }

    turbo::Result<int64_t> SlowRandomAccessFile::read(int64_t nbytes, void *out) {
        latencies_->Sleep();
        return stream_->read(nbytes, out);
    }

    turbo::Result<std::shared_ptr<Buffer>> SlowRandomAccessFile::read(int64_t nbytes) {
        latencies_->Sleep();
        return stream_->read(nbytes);
    }

    turbo::Result<int64_t> SlowRandomAccessFile::read_at(int64_t position, int64_t nbytes,
                                                 void *out) {
        latencies_->Sleep();
        return stream_->read_at(position, nbytes, out);
    }

    turbo::Result<std::shared_ptr<Buffer>> SlowRandomAccessFile::read_at(int64_t position,
                                                                 int64_t nbytes) {
        latencies_->Sleep();
        return stream_->read_at(position, nbytes);
    }

    turbo::Result<std::string_view> SlowRandomAccessFile::peek(int64_t nbytes) {
        return stream_->peek(nbytes);
    }

}  // namespace nebula::io

